/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package aiaudio.processing.network.manager;

import aiaudio.processing.network.network.DBAudioNetwork;
import aiaudio.processing.base.helpers.Algorithmic;
import aiaudio.processing.base.mapreduce.CoordinateDoubleValueKey;
import aiaudio.processing.base.mapreduce.standard.mappers.UserToArtistValueMatrixMapper;
import aiaudio.lastfm.hbase.CannotCreateTableException;
import aiaudio.Application;
import aiaudio.database.tables.DataMetricsTable;
import aiaudio.database.tables.RatingsMatrixTable;
import aiaudio.database.tables.ReduceDictionaryTable;
import aiaudio.processing.MainTableGroup;
import java.io.IOException;
import java.util.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.mapreduce.Job;

/**
 *
 * @author nastya
 */
public class DBAudioNetworkManager {

    private DBAudioNetwork network;
    private int learningCicles;
    private double divisor;
    private double minSimilarity;
    private int trainingSetSize;
    private String temporaryTableName;
    private final MainTableGroup mainTableGroup;

    public void setDivisor(double divisor) {
        this.divisor = divisor;
    }

    public void setLearningCicles(int learningCicles) {
        this.learningCicles = learningCicles;
    }

    public void setMinSimilarity(double minSimilarity) {
        this.minSimilarity = minSimilarity;
    }

    public void setTrainingSetSize(int trainingSetSize) {
        this.trainingSetSize = trainingSetSize;
    }

    public DBAudioNetworkManager(MainTableGroup mainTableGroup) {
        this.mainTableGroup = mainTableGroup;
    }

    protected HashMap<String, double[]> collectTrainingRatings(List<String> userKeys, Vector<String> artists) throws IOException {
        HashMap<String, double[]> r = createEmptyMap(userKeys, artists.size());
        HTable ratingsTable = Application.database().getByName(mainTableGroup.getPrepareRatingsAlgTableGroup().getNormilizedRatingMatrix()).getTable();
        fillTrainingRatings(ratingsTable, artists, userKeys, r);
        return r;
    }

    protected void fillTrainingRatings(HTable ratingsTable, Vector<String> artists, List<String> userKeys, HashMap<String, double[]> r) throws IOException {
        Result defaultVotingResult = RatingsMatrixTable.getDefaultRatings(ratingsTable);
        Scan scan = new Scan();
        ResultScanner scanner = ratingsTable.getScanner(scan);
        Result result;
        while ((result = scanner.next()) != null) {
            byte[] artistKey = result.getRow();
            String artistKeyString = Bytes.toString(artistKey);

            if (!artists.contains(artistKeyString)) {
                continue;
            }
            processArtistRow(userKeys, result, defaultVotingResult, r);
        }
    }

    protected void makeGroupping(Vector<String> artists, HTable outputTable) throws IOException {
        HTable table = Application.database().getByName(temporaryTableName).getTable();
        Scan scan = new Scan();
        ResultScanner scanner = table.getScanner(scan);
        Result res;
        while ((res = scanner.next()) != null) {
            byte[] userKey = res.getRow();
            double[] ratings = new double[artists.size()];
            for (int i = 0; i < artists.size(); i++) {
                ratings[i] = RatingsMatrixTable.getRatingByArtist(res, artists.get(i));
            }
            network.saveGroupping(outputTable, ratings, userKey);
        }
    }

    protected void processArtistRow(List<String> userKeys, Result result, Result defaultVotingResult, HashMap<String, double[]> r) {
        for (String userKey : userKeys) {
            processUserColumnInArtistRow(result, userKey, defaultVotingResult, userKeys, r);
        }
    }

    protected void processUserColumnInArtistRow(Result result, String userKey, Result defaultVotingResult, List<String> userKeys, HashMap<String, double[]> r) {
        double rat = RatingsMatrixTable.getArtistRatingOrDefault(result, defaultVotingResult, Bytes.toBytes(userKey));
        int idx = userKeys.indexOf(userKey);
        r.get(userKey)[idx] = rat;
    }

    protected HashMap<String, double[]> createEmptyMap(List<String> userKeys, int artistsCount) {
        HashMap<String, double[]> r = new HashMap<String, double[]>();
        for (String s : userKeys) {
            r.put(s, new double[artistsCount]);
        }
        return r;
    }

    protected List<String> collectUserKeys(Vector<Integer> indices) throws IOException {
        //TODO use map reduce?
        HTable usersTable = Application.database().getByName(mainTableGroup.getInitialDataTableGroup().getUsersTable()).getTable();
        List<String> userKeys = new Vector<String>();
        Scan scan = new Scan();
        ResultScanner scanner = usersTable.getScanner(scan);
        Result result;
        int count = 0;
        while ((result = scanner.next()) != null) {
            if (indices.contains(count)) {
                byte[] row = result.getRow();
                userKeys.add(Bytes.toString(row));
            }
            count++;
        }
        scanner.close();
        return userKeys;
    }
    public Vector<String> artists;

    public void start() throws IOException, CannotCreateTableException, InterruptedException, ClassNotFoundException {
        temporaryTableName = Application.database().createTemporaryMatrixTable();
        network = new DBAudioNetwork(learningCicles, minSimilarity);


        HTable outputTable = Application.database().getByName(mainTableGroup.getNetworkAlgTableGroup().getUserToGroupTable()).getTable();

        artists = collectNeuronArtists();
        System.out.println("collected artists count: " + artists.size());

        HashMap<String, double[]> trainingRatings = collectTrainingRatings(trainingSetSize, artists);
        System.out.println("collected users count: " + trainingRatings.entrySet().size());

        int tracksCount = artists.size();
        int usersCount = trainingRatings.keySet().size();

        network.configure(usersCount, tracksCount, divisor);
        System.out.println("configured");

        Iterator<double[]> iterator = trainingRatings.values().iterator();
        network.train(iterator);
        System.out.println("trained");

        Job job = createPrepareToGrouppingJob(artists);
        job.waitForCompletion(true);

        makeGroupping(artists, outputTable);
        System.out.println("groupping ended");
    }

    protected HashMap<String, double[]> collectTrainingRatings(int trainingSetSize, Vector<String> artistKeyList) throws IOException {
        Vector<Integer> indices = Algorithmic.randomValues(trainingSetSize, calculateTotalUsersCount());
        List<String> userKeyList = collectUserKeys(indices);
        HashMap<String, double[]> r = collectTrainingRatings(userKeyList, artistKeyList);
        return r;
    }

    private Vector<String> collectNeuronArtists() throws IOException {
        Vector<String> res = new Vector<String>();

        HTable artistsTable = Application.database().getByName(mainTableGroup.getInitialDataTableGroup().getArtistsTable()).getTable();
        Scan scan = new Scan();
        ResultScanner scanner = artistsTable.getScanner(scan);
        Result result;
        while ((result = scanner.next()) != null) {
            NavigableMap<byte[], byte[]> familyMap = ReduceDictionaryTable.getCentroids(result);
            for (byte[] bs : familyMap.values()) {
                String artist = Bytes.toString(bs);
                if (!res.contains(artist)) {
                    res.add(artist);
                }
            }
        }
        scanner.close();
        return res;
    }

    private int calculateTotalUsersCount() throws IOException {
        HTable dataMetricsTable = Application.database().getByName(mainTableGroup.getInitialDataTableGroup().getDataMetricsTable()).getTable();
        return (int) DataMetricsTable.getUsersCount(dataMetricsTable);
    }

    private Job createPrepareToGrouppingJob(Vector<String> artists) throws IOException {
        Job job = new Job();

        Scan scan = new Scan();
        
        job.getConfiguration().set(CollectNecessaryValuesReducer.RATING_MATRIX_NAME, mainTableGroup.getPrepareRatingsAlgTableGroup().getNormilizedRatingMatrix());
        job.getConfiguration().setStrings(CollectNecessaryValuesReducer.INCLUDE_ARTISTS, artists.toArray(new String[0]));

        String ratingsTableNameDb = Application.database().getByName(mainTableGroup.getPrepareRatingsAlgTableGroup().getNormilizedRatingMatrix()).getDatabaseName();
        TableMapReduceUtil.initTableMapperJob(ratingsTableNameDb, scan, UserToArtistValueMatrixMapper.class, ImmutableBytesWritable.class, CoordinateDoubleValueKey.class, job);
        
        String temporaryTableDBName = Application.database().getByName(temporaryTableName).getDatabaseName();
        TableMapReduceUtil.initTableReducerJob(temporaryTableDBName, CollectNecessaryValuesReducer.class, job);
        return job;

    }
}
